package com.zhanghe.study.hbase;


import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/**
 * Created by cschen on 16/11/17.
 */
public class UserBiz {
    private final Logger LOGGER = LoggerFactory.getLogger(UserBiz.class);

    private static Connection connection;
    static {
        try {
            // 获取连接
            connection = ConnectionFactory.createConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void addExpose(String uid, String campId){
        long ts = System.currentTimeMillis();
        List<Long> tsList;
        try {
            tsList = getExposeFromHbase(uid, campId);
        } catch (Exception e) {
            LOGGER.error(e.toString());
            return;
        }
        if (tsList == null) { // 之前没有
            tsList = new ArrayList<>();
        }
        tsList.add(ts);
        // 添加曝光到hbase中
        addExposeInHbase(uid, campId,tsList);
    }

    /**
     * 添加一次曝光到Hbase
     *
     * @param uid
     * @param campId
     * @param tsList
     */
    public void addExposeInHbase(String uid, String campId,List<Long> tsList) {

        Table htable = null;
        try {
            htable = connection.getTable(TableName.valueOf("expose"));
            Gson gson = new Gson();
            Put put = new Put(Bytes.toBytes(uid));
            put.addColumn(Bytes.toBytes("f"), Bytes.toBytes(campId), Bytes.toBytes(gson.toJson(tsList)));
            //增加过期时间  90天
            put.setTTL(3600 * 24 * 90 * 1000L);
            htable.put(put);
        } catch (IOException e) {
            LOGGER.error("write to hbase failed: {}", uid + ":" + e.getMessage());
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (Exception closeExcption) {
                    LOGGER.error("close connection failed: ", closeExcption);
                }
            }
        }
    }

    // 是否超频
    private boolean isOver(String uid, String campId, long range, int limit) {

        long cur = System.currentTimeMillis();
        List<Long> exposes = getExposeFromHbase(uid, campId);

        int count = 0;
        if (exposes == null || exposes.isEmpty()) {
            return false;
        } else {
            for (long time : exposes) {
                if (time > cur - range) {
                    count++;
                } else {
                    break;
                }
            }
        }
        return count > limit;
    }

    /**
     * 从Hbase中获取到指定Camp的曝光列表
     *
     * @param uid
     * @param campId
     * @return
     */
    public List<Long> getExposeFromHbase(String uid, String campId) {
        if (StringUtils.isBlank(uid) || StringUtils.isBlank(campId)) {
            return null;
        }

        Table htable = null;
        try {
            // 对应的表
            htable = connection.getTable(TableName.valueOf("expose"));
            // 用户id作为rowkey
            Get get = new Get(Bytes.toBytes(uid));
            // 列族 列名
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(campId));

            Result result = htable.get(get);
            if (result != null && !result.isEmpty()) {
                // 内容
                byte[] value = result.getValue(Bytes.toBytes("f"), Bytes.toBytes(campId));

                String s = new String(value);
                Gson gson = new Gson();
                final String creative = gson.fromJson(s, String.class);
                // 历史曝光的时间
                List<Long> tsList = gson.fromJson(s, new TypeToken<ArrayList<Long>>() {
                }.getType());
                return tsList;
            }
        } catch (IOException e) {
            LOGGER.error("get camp list failed: {}", uid);
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (Exception closeExcption) {
                    LOGGER.error("close connection failed: ", closeExcption);
                }
            }
        }
        return null;
    }

    public static void main(String[] args) {
        UserBiz userBiz = new UserBiz();
//        userBiz.isOver("uid1","101",3600*1000L*24,3);
        userBiz.addExpose("uid1","110");
    }
}
